using System;
using System.Collections.Generic;
using Me.Prettyprint.Cassandra.Model;
using Me.Prettyprint.Cassandra.Model.Thrift;
using Me.Prettyprint.Cassandra.Service;
using Me.Prettyprint.Cassandra.Utils;
using Me.Prettyprint.Hector.Api;
using Me.Prettyprint.Hector.Api.Beans;
using Me.Prettyprint.Hector.Api.Factory;
using Me.Prettyprint.Hector.Api.Mutation;
using Me.Prettyprint.Hector.Api.Query;
using Org.Apache.Cassandra.Thrift;
using Thrift;

namespace Me.Prettyprint.Cassandra.Service.Template
{
	public class ThriftSuperCfTemplate<K, SN, N> : SuperCfTemplate<K, SN, N>
	{
        public ThriftSuperCfTemplate(Keyspace keyspace, String columnFamily,
            Serializer<K> keySerializer, Serializer<SN> topSerializer,
            Serializer<N> subSerializer)
            : base(keyspace, columnFamily, keySerializer, topSerializer, subSerializer)
        {
        }
        protected override SuperCfResult<K, SN, N> DoExecuteSlice(K key, SN sColumnName, HSlicePredicate<SN> predicate)
        {
            SuperCfResultWrapper<K, SN, N> wrapper =
              new SuperCfResultWrapper<K, SN, N>(KeySerializer, TopSerializer, SubSerializer,
                  SliceInternal(key, predicate));
            if (sColumnName != null)
            {
                wrapper.ApplySuperColumn(sColumnName);
            }
            return wrapper;
        }
        protected override SuperCfResult<K, SN, N> DoExecuteMultigetSlice(List<K> keys, HSlicePredicate<SN> predicate)
        {
            SuperCfResultWrapper<K, SN, N> wrapper =
              new SuperCfResultWrapper<K, SN, N>(KeySerializer, TopSerializer, SubSerializer,
                  MultigetSliceInternal(keys, ColumnParent, predicate));
            return wrapper;
        }
        private ExecutionResult<Dictionary<byte[], List<ColumnOrSuperColumn>>> SliceInternal(K key, HSlicePredicate<SN> workingSlicePredicate)
        {
            Func<Org.Apache.Cassandra.Thrift.Cassandra.Client, Dictionary<byte[], List<ColumnOrSuperColumn>>> x = (Org.Apache.Cassandra.Thrift.Cassandra.Client cassandra) =>
            {
                try
                {
                    Dictionary<byte[],List<ColumnOrSuperColumn>> cosc = new Dictionary<byte[], List<ColumnOrSuperColumn>>();
                    byte[] sKey = KeySerializer.ToBytes(key);
                    cosc[sKey] =  cassandra.get_slice(sKey, ColumnParent, workingSlicePredicate.ToThrift(), 
                        ThriftConverter.consistencyLevel(ConsistencyLevelPolicy.Get(OperationType.READ)));
                    return cosc;
                }
                catch (Exception e)
                {
                    throw ExceptionsTranslator.translate(e);
                }
            };
            Operation<Dictionary<byte[], List<ColumnOrSuperColumn>>> op = new OperationBuilder<Dictionary<byte[], List<ColumnOrSuperColumn>>>(x);
            op.OperationType = OperationType.READ;
            return ((ExecutingKeyspace)Keyspace).DoExecuteOperation(op);
        }
        private ExecutionResult<Dictionary<byte[], List<ColumnOrSuperColumn>>> MultigetSliceInternal(List<K> keys,
            ColumnParent workingColumnParent, HSlicePredicate<SN> workingSlicePredicate)
        {
            Func<Org.Apache.Cassandra.Thrift.Cassandra.Client, Dictionary<byte[], List<ColumnOrSuperColumn>>> x = (Org.Apache.Cassandra.Thrift.Cassandra.Client cassandra) =>
            {
                try
                {
                    List<byte[]> sKeys = KeySerializer.ToBytesList(keys);
                    Dictionary<byte[], List<ColumnOrSuperColumn>> cosc = cassandra.multiget_slice(sKeys, workingColumnParent,
                        workingSlicePredicate.ToThrift(), 
                        ThriftConverter.consistencyLevel(ConsistencyLevelPolicy.Get(OperationType.READ)));
                    return cosc;
                }
                catch (Exception e)
                {
                    throw ExceptionsTranslator.translate(e);
                }
            };
            Operation<Dictionary<byte[], List<ColumnOrSuperColumn>>> op = new OperationBuilder<Dictionary<byte[], List<ColumnOrSuperColumn>>>(x);
            op.OperationType = OperationType.READ;
            return ((ExecutingKeyspace)Keyspace).DoExecuteOperation(op);
        }
	}
}